package com.uptown.matrix.core.consumer.kafka;


import java.lang.reflect.Method;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

import javax.annotation.Resource;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import com.uptown.matrix.core.annotation.MatrixMessageListener;

@Slf4j
@Component
public class KafkaConsumerHandler implements DisposableBean {

    @Resource
    Map<String, KafkaConsumerRunnable> kafkaConsumerMap;


    public void createSubscription(Object bean, MatrixMessageListener annotation, Method method, String groupTopicKey) throws Throwable {
        String topic = annotation.streamKey();
        String groupId = annotation.consumerGroup();

        Properties props = new Properties();
        props.put("bootstrap.servers", annotation.bootServer());
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "false");
        props.put("auto.offset.reset", "latest");
        props.put("max.poll.records", 5);
        props.put("session.timeout.ms", "60000");
        props.put("max.poll.interval.ms", 300000);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  //键反序列化方式
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        consumer.subscribe(Collections.singletonList(topic));

        log.info("init kafka consumer -> {}", topic);
        KafkaConsumerRunnable kafkaConsumerRunnable = new KafkaConsumerRunnable(consumer, bean, method);
        kafkaConsumerMap.put(groupTopicKey, kafkaConsumerRunnable);

        // 启动监听
        new Thread(kafkaConsumerRunnable, groupTopicKey).start();
    }


    @Override
    public void destroy() {
        kafkaConsumerMap.forEach((topic, task) -> {
            task.shutDown();
        });
    }

    @Scheduled(cron = "*/30 * * * * ?")
    public void kafkaConsumerCheck() {
        Thread mainThread = Thread.currentThread();
        ThreadGroup mainThreadGroup = mainThread.getThreadGroup();

        int count = mainThreadGroup.activeCount();
        Thread[] threads = new Thread[count];

        mainThreadGroup.enumerate(threads, true);
        List<String> threadNameList = Arrays.stream(threads).map(Thread::getName).collect(Collectors.toList());

        kafkaConsumerMap.forEach((groupTopicKey, task) -> {

            if (!threadNameList.contains(groupTopicKey)) {

                new Thread(task, groupTopicKey).start();
            }
        });
    }
}

